#include "zmq.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

static size_t message_size = 1000;
static int roundtrip_count = 1000;

static void worker(void* ctx_)
{
    void* s;
    int rc;
    int i;
    zmq_msg_t msg;

    //s = zmq_socket (ctx_, ZMQ_REP);
    s = zmq_socket(ctx_, ZMQ_PAIR);
    if (!s) {
        exit(1);
    }

    rc = zmq_connect(s, "inproc://lat_test");
    if (rc != 0) {
        exit(1);
    }
    int timeout = 2000;
    zmq_setsockopt(s, ZMQ_RCVTIMEO, &timeout, sizeof(int));

    rc = zmq_msg_init(&msg);
    if (rc != 0) {
        exit(1);
    }

#if 0
    for (i = 0; i != roundtrip_count; i++) {
        rc = zmq_recvmsg(s, &msg, 0);
        if (rc < 0) {
            continue;
        }
        rc = zmq_sendmsg(s, &msg, 0);
        if (rc < 0) {
            exit(1);
        }
    }
#endif    
    zmq_pollitem_t items[] = {
        { s, 0, ZMQ_POLLIN, 0 },
    };
    
    //while (1) {
    for (i = 0; i != roundtrip_count; i++) {
        zmq_poll(items, 1, 2000);
        if (items[0].revents & ZMQ_POLLIN) {
            //zmq_msg_init(&msg);
            rc = zmq_recvmsg(s, &msg, 0);
            if (rc < 0) {
                continue;
            }
            rc = zmq_sendmsg(s, &msg, 0);
            if (rc < 0) {
                exit(1);
            }
            //zmq_msg_close(&msg);
            printf("++++++++++++++++++++++++++++%d\r\n", i);
        }
        else {
            printf("++++++++++++++++++++++++++++%d\r\n", i);
        }
    }
    
    rc = zmq_msg_close(&msg);
    if (rc != 0) {
        exit(1);
    }

    rc = zmq_close(s);
}

int main_testinproc()
{
    void* ctx;
    void* s;
    int rc;
    int i;
    zmq_msg_t msg;
    void* watch;
    unsigned long elapsed;
    double latency;

    ctx = zmq_init(1);
    if (!ctx) {
        return -1;
    }

    //s = zmq_socket (ctx, ZMQ_REQ);
    s = zmq_socket(ctx, ZMQ_PAIR);
    if (!s) {
        return -1;
    }

    rc = zmq_bind(s, "inproc://lat_test");
    if (rc != 0) {
        return -1;
    }

    zmq_threadstart(worker, ctx);
    if (rc != 0) {
        return -1;
    }

    rc = zmq_msg_init_size(&msg, message_size);
    if (rc != 0) {
        return -1;
    }
    memset(zmq_msg_data(&msg), 0, message_size);

    printf("message size: %d [B]\n", (int)message_size);
    printf("roundtrip count: %d\n", (int)roundtrip_count);

    watch = zmq_stopwatch_start();

    for (i = 0; i != roundtrip_count; i++) {
        //zmq_msg_init_size(&msg, message_size);
        //memset(zmq_msg_data(&msg), 0, message_size);

        rc = zmq_sendmsg(s, &msg, 0);

        if (rc < 0) {
            return -1;
        }

        rc = zmq_recvmsg(s, &msg, 0);
        if (rc < 0) {
            return -1;
        }
        if (zmq_msg_size(&msg) != message_size) {
            return -1;
        }
    }

    elapsed = zmq_stopwatch_stop(watch);

    rc = zmq_msg_close(&msg);
    if (rc != 0) {
        return -1;
    }

    latency = (double)elapsed / (roundtrip_count * 2);
    printf("average latency: %.3f [us]\n", (double)latency);

    rc = zmq_close(s);
    if (rc != 0) {
        return -1;
    }

    rc = zmq_ctx_term(ctx);
    if (rc != 0) {
        return -1;
    }

    return 0;
}
